package com.deng.eventflow.core;

import com.deng.eventflow.core.exception.ProcessorException;
import com.deng.eventflow.core.exception.Variant;
//import com.deng.eventflow.core.io.Printable;
//import com.deng.eventflow.core.io.Readable;
import com.deng.eventflow.core.petitpoucet.NodeFunction;
import com.deng.eventflow.core.petitpoucet.ProvenanceNode;


import java.util.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 接收零个或多个输入事件，并产生零个或多个输出事件。
 *
 * 处理器是所有计算发生的基础类。 eventflow 的所有处理器（包括你的）都是这个类的后代。
 *
 * 处理器在图形上被描述为一个“盒子”，“管道”代表它的输入和输出流。
 *
 * 这个类本身是抽象的； 然而，它提供了处理输入/输出事件队列、将处理器连接在一起等的重要方法。
 * 但是，如果您编写自己的处理器，您很可能希望从它的子处理器 SynchronousProcessor 继承，它会为您做更多的工作。
 *
 * Processor 类不假定输入或输出事件的类型。 因此，它的所有输入和输出队列都被声明为包含 Java 最通用类型 Object 的实例。
 */

public abstract class Processor implements DuplicableProcessor,
        Contextualizable {

    protected int m_inputArity;

    protected int m_outputArity;

    public static final transient String s_versionString = "0.10.5";

    protected transient Queue<Object>[] m_inputQueues;

    protected transient EventTracker m_eventTracker = null;

    protected transient Queue<Object>[] m_outputQueues;

    protected transient Pullable[] m_inputPullables;

    protected transient Pushable[] m_outputPushables;

    protected int m_inputCount = 0;

    protected int m_outputCount = 0;

    private static int s_uniqueIdCounter = 0;

    private static transient Lock s_counterLock = new ReentrantLock();

    private final int m_uniqueId;

    protected Context m_context = null;

    public static final transient int MAX_PULL_RETRIES = 10000000;

    protected boolean[] m_hasBeenNotifiedOfEndOfTrace;

    /**
     * 初始化处理器。
     *
     * 这具有执行每个处理器共有的基本操作的效果：
     *
     * 提供唯一 ID
     * 确定其输入和输出数量
     * 创建空输入和输出队列数组，以及 Pullables 和 Pushables 数组
     * 如果您创建自己的处理器，它的构造函数必须从调用其祖先的构造函数开始（最终调用此构造函数）。否则，会出现大量的空指针和其他奇怪的东西。
     * @param in_arity 处理器的输入参数
     * @param out_arity 处理器的输出数量
     */
    public Processor(int in_arity, int out_arity) {
        super();
        m_inputArity = in_arity;
        m_outputArity = out_arity;
        s_counterLock.lock();
        m_uniqueId = s_uniqueIdCounter++;
        s_counterLock.unlock();
        m_inputQueues = new Queue[m_inputArity];
        for (int i = 0; i < m_inputArity; i++)
        {
            m_inputQueues[i] = new ArrayDeque<Object>();
        }
        m_outputQueues = new Queue[m_outputArity];
        for (int i = 0; i < m_outputArity; i++)
        {
            m_outputQueues[i] = new ArrayDeque<Object>();
        }
        m_inputPullables = new Pullable[m_inputArity];
        m_outputPushables = new Pushable[m_outputArity];
        m_hasBeenNotifiedOfEndOfTrace = new boolean[m_inputArity];
        for (int i = 0; i < m_inputArity; i++)
        {
            m_hasBeenNotifiedOfEndOfTrace[i] = false;
        }

    }

    protected boolean allNotifiedEndOfTrace() {
        for (int i = 0; i < m_inputArity; i++)
        {
            if (!m_hasBeenNotifiedOfEndOfTrace[i])
            {
                return false;
            }
        }
        return true;
    }

    protected final Context newContext () {
        return new Context();
    }

    public final synchronized Object getContext(String key) {
        if (m_context == null || !m_context.containsKey(key))
        {
            return null;
        }
        return m_context.get(key);

    }

    public synchronized  Context getContext() {
        if (m_context == null)
        {
            m_context = newContext();
        }
        return m_context;
    }

//    @Override
    public synchronized void setContext(String key, Object value) {
        if (m_context == null)
        {
            m_context = newContext();
        }
        m_context.put(key, value);
    }

    @Override
    public synchronized void setContext(Context context) {
        if (context != null)
        {
            if (m_context == null)
            {
                m_context = newContext();
            }
            m_context.putAll(context);
        }
    }

    @Override
    public final int hashCode() {
        return m_uniqueId;
    }

    @Override
    public final boolean equals(Object o) {
        if (o == null || !(o instanceof Processor))
        {
            return false;
        }
        Processor p = (Processor) o;
        return m_uniqueId == p.m_uniqueId;
    }

    public final int getId() {
        return m_uniqueId;
    }

    public synchronized void reset() {
        // Reset input
        for (int i = 0; i < m_inputArity; i++)
        {
            m_inputQueues[i].clear();
        }
        // Reset output
        for (int i = 0; i < m_outputArity; i++)
        {
            m_outputQueues[i].clear();
        }
        for (int i = 0; i < m_inputArity; i++)
        {
            m_hasBeenNotifiedOfEndOfTrace[i] = false;
        }
        m_inputCount = 0;
        m_outputCount = 0;
    }

    public abstract Pushable getPushableInput(int index);

    public final synchronized Pushable getPushableInput() {
        return getPushableInput(0);
    }

    public abstract Pullable getPullableOutput(int index);

    public final synchronized Pullable getPullableOutput() {
        return getPullableOutput(0);
    }

    public synchronized void setPullableInput(int i, Pullable p) {
        m_inputPullables[i] = p;
    }

    public synchronized Pullable getPullableInput(int i) {
        return m_inputPullables[i];
    }

    public synchronized void setPushableOutput(int i, Pushable p) {
        m_outputPushables[i] = p;
    }

    public synchronized Pushable getPushableOutput(int i) {
        return m_outputPushables[i];
    }

    public final int getInputArity() {
        return m_inputArity;
    }

    public final int getOutputArity() {
        return m_outputArity;
    }

    public static boolean allNull(Object[] v) {
        for (Object o : v)
        {
            if (o != null)
            {
                return false;
            }
        }
        return true;
    }

    public void duplicateInto(Processor p) {
        p.m_eventTracker = m_eventTracker;
        p.setContext(m_context);
        for (int i = 0; i < m_inputQueues.length; i++)
        {
            p.m_inputQueues[i].addAll(m_inputQueues[i]);
        }
        for (int i = 0; i < m_outputQueues.length; i++)
        {
            p.m_outputQueues[i].addAll(m_outputQueues[i]);
        }
    }

    public final Set<Class<?>> getInputType(int index) {
        Set<Class<?>> classes = new HashSet<Class<?>>();
        if (index >= 0 && index < m_inputArity)
        {
            getInputTypesFor(classes, index);
        }
        return classes;
    }


    public void getInputTypesFor(Set<Class<?>> classes, int index) {
        classes.add(Variant.class);
    }

    public Class<?> getOutputType(int index) {
        return Variant.class;
    }

    public static Queue<Object[]> getEmptyQueue() {
        return new ArrayDeque<Object[]>();
    }

    public void start() {
        // Nothing
    }

    public void stop() {
        // Nothing
    }

    public static void startAll(Processor ... procs) {
        for (Processor p : procs)
        {
            if (p != null)
            {
                p.start();
            }
        }
    }

    public static void stopAll(Processor ... procs) {
        for (Processor p : procs)
        {
            if (p != null)
            {
                p.stop();
            }
        }
    }

    public final EventTracker getEventTracker() {
        return m_eventTracker;
    }

    public Processor setEventTracker(EventTracker tracker) {
        m_eventTracker = tracker;
        return this;
    }

    public void associateToInput(int in_stream_index, int in_stream_pos, int out_stream_index,
                                 int out_stream_pos) {
        if (m_eventTracker != null)
        {
            m_eventTracker.associateToInput(m_uniqueId, in_stream_index, in_stream_pos, out_stream_index,
                    out_stream_pos);
        }
    }

    public void associateTo(NodeFunction f, int out_stream_index, int out_stream_pos) {
        if (m_eventTracker != null)
        {
            m_eventTracker.associateTo(m_uniqueId, f, out_stream_index, out_stream_pos);
        }
    }

    public void associateToOutput(int in_stream_index, int in_stream_pos, int out_stream_index,
                                  int out_stream_pos) {
        if (m_eventTracker != null)
        {
            m_eventTracker.associateToOutput(m_uniqueId, in_stream_index, in_stream_pos, out_stream_index,
                    out_stream_pos);
        }
    }

    protected boolean onEndOfTrace(Queue<Object[]> outputs) throws ProcessorException {
        return false;
    }

    public final int getInputCount() {
        return m_inputCount;
    }

    public final int getOutputCount() {
        return m_outputCount;
    }

//    public final Object print(ObjectPrinter<?> printer) throws ProcessorException {
//        Map<String,Object> contents = new HashMap<String,Object>();
//        contents.put("id", m_uniqueId);
//        contents.put("input-count", m_inputCount);
//        contents.put("output-count", m_outputCount);
//        contents.put("context", m_context);
//        List<Queue<Object>> in_queues = new ArrayList<Queue<Object>>(m_inputQueues.length);
//        for (Queue<Object> q : m_inputQueues)
//        {
//            in_queues.add(q);
//        }
//        contents.put("input-queues", in_queues);
//        List<Queue<Object>> out_queues = new ArrayList<Queue<Object>>(m_outputQueues.length);
//        for (Queue<Object> q : m_outputQueues)
//        {
//            out_queues.add(q);
//        }
//        contents.put("output-queues", out_queues);
//        contents.put("contents", printState());
//        try
//        {
//            return printer.print(contents);
//        }
//        catch (PrintException e)
//        {
//            throw new ProcessorException(e);
//        }
//    }

    protected Object printState() {
        return null;
    }

//    public final Processor read(ObjectReader<?> reader, Object o) throws ProcessorException {
//        Map<String, Object> contents = null;
//        try
//        {
//            contents = (Map<String,Object>) reader.read(o);
//        }
//        catch (ReadException e)
//        {
//            throw new ProcessorException(e);
//        }
//        Processor p = null;
//        if (contents.containsKey("contents"))
//        {
//            Object o_contents = contents.get("contents");
//            try
//            {
//                p = readState(o_contents);
//            }
//            catch (UnsupportedOperationException e)
//            {
//                throw new ProcessorException(e);
//            }
//        }
//        if (p == null)
//        {
//            throw new ProcessorException("The processor returned null with being deserialized");
//        }
//        p.m_inputCount = ((Number) contents.get("input-count")).intValue();
//        p.m_outputCount = ((Number) contents.get("output-count")).intValue();
//        try
//        {
//            reader.setField(p, "m_uniqueId", ((Number) contents.get("id")).intValue());
//        }
//        catch (ReadException e)
//        {
//            throw new ProcessorException(e);
//        }
//        List<Queue<Object>> in_queues = (List<Queue<Object>>) contents.get("input-queues");
//        for (int i = 0; i < in_queues.size(); i++)
//        {
//            p.m_inputQueues[i] = in_queues.get(i);
//        }
//        List<Queue<Object>> out_queues = (List<Queue<Object>>) contents.get("output-queues");
//        for (int i = 0; i < out_queues.size(); i++)
//        {
//            p.m_outputQueues[i] = in_queues.get(i);
//        }
//        return p;
//    }

    protected Processor readState(Object o) {
        throw new UnsupportedOperationException("This processor does not support deserialization");
    }

    public final Processor duplicate() {
        return duplicate(false);
    }

    public abstract Processor duplicate(boolean with_state);

    public static List<ProvenanceNode> getLeaves(ProvenanceNode root) {
        List<ProvenanceNode> leaves = new ArrayList<ProvenanceNode>();
        getLeaves(root, leaves);
        return leaves;
    }

    protected static void getLeaves(ProvenanceNode root, List<ProvenanceNode> leaves) {
        if (root == null)
        {
            return;
        }
        List<ProvenanceNode> children = root.getChildren();
        if (children.isEmpty())
        {
            leaves.add(root);
        }
        else
        {
            for (ProvenanceNode child : children)
            {
                getLeaves(child, leaves);
            }
        }
    }

}
